<?php

namespace Swoole\Client;

use Swoole\Protocol\SOAServer;

class SOA {

	protected $servers = array();

	protected $wait_list = array();

	protected $timeout = 0.5;

	protected $packet_maxlen = 2097152;
	// 最大不超过2M的数据包
	const OK = 0;

	const TYPE_ASYNC = 1;

	const TYPE_SYNC = 2;

	public $re_connect = true;
	// 重新connect
	
	/**
	 * 发送请求
	 * @param $type
	 * @param $send
	 * @param $retObj
	 */
	protected function request($type, $send, $retObj) {
		$socket = new \Swoole\Client\TCP();
		$retObj->socket = $socket;
		$retObj->type = $type;
		$retObj->send = $send;
		
		$svr = $this->getServer();
		// 异步connect
		$ret = $socket->connect($svr['host'], $svr['port'], $this->timeout);
		// 使用SOCKET的编号作为ID
		$retObj->id = (int)$socket->get_socket();
		if ($ret === false) {
			$retObj->code = SOA_Result::ERR_CONNECT;
			unset($retObj->socket);
			return false;
		}
		// 发送失败了
		if ($retObj->socket->send(SOAServer::encode($retObj->send)) === false) {
			$retObj->code = SOA_Result::ERR_SEND;
			unset($retObj->socket);
			return false;
		}
		// 加入wait_list
		if ($type != self::TYPE_ASYNC) {
			$this->wait_list[$retObj->id] = $retObj;
		}
		return true;
	}

	/**
	 * 完成请求
	 * @param $retData
	 * @param $retObj
	 */
	protected function finish($retData, $retObj) {
		// 解包失败了
		if ($retData === false) {
			$retObj->code = SOA_Result::ERR_UNPACK;
		} elseif ($retData['errno'] === self::OK) { // 调用成功
			$retObj->code = self::OK;
			$retObj->data = $retData['data'];
		} else { // 服务器返回失败
			$retObj->code = SOA_Result::ERR_SERVER;
			$retObj->data = '';
		}
		if ($retObj->type != self::TYPE_ASYNC) {
			unset($this->wait_list[$retObj->id]);
		}
		if ($retObj->callback) {
			call_user_func($retObj->callback, $retObj);
		}
	}

	/**
	 * 添加服务器
	 * @param array $servers
	 */
	public function addServers(array $servers) {
		if (isset($servers['host'])) {
			$this->servers[] = $servers;
		} else {
			$this->servers = array_merge($this->servers, $servers);
		}
	}

	/**
	 * 从配置中取出一个服务器配置
	 * @return array
	 * @throws \Exception
	 */
	public function getServer() {
		if (empty($this->servers)) {
			throw new \Exception("servers config empty.");
		}
		$_svr = $this->servers[array_rand($this->servers)];
		$svr = array(
				'host' => '','port' => 0 
		);
		list($svr['host'], $svr['port']) = explode(':', $_svr, 2);
		return $svr;
	}

	/**
	 * RPC调用
	 *
	 * @param $function
	 * @param $params
	 * @param $callback
	 * @return SOA_Result
	 */
	public function task($function, $params = array(), $callback = null) {
		$retObj = new SOA_Result();
		$send = array(
				'call' => $function,'params' => $params 
		);
		$this->request(self::TYPE_SYNC, $send, $retObj);
		$retObj->callback = $callback;
		return $retObj;
	}

	/**
	 * 异步任务
	 * @param $function
	 * @param $params
	 * @return SOA_Result
	 */
	public function async($function, $params) {
		$retObj = new SOA_Result();
		$send = array(
				'call' => $function,'params' => $params 
		);
		$this->request(self::TYPE_ASYNC, $send, $retObj);
		if ($retObj->socket != null) {
			$recv = $retObj->socket->recv();
			if ($recv == false) {
				$retObj->code = SOA_Result::ERR_TIMEOUT;
				return $retObj;
			}
			$this->finish(SOAServer::decode($recv), $retObj);
		}
		return $retObj;
	}

	/**
	 * 并发请求
	 * @param float $timeout
	 * @return int
	 */
	public function wait($timeout = 0.5) {
		$st = microtime(true);
		$t_sec = (int)$timeout;
		$t_usec = (int)(($timeout - $t_sec) * 1000 * 1000);
		$buffer = $header = array();
		$success_num = 0;
		
		while (true) {
			$write = $error = $read = array();
			if (empty($this->wait_list)) {
				break;
			}
			foreach ($this->wait_list as $obj) {
				if ($obj->socket !== null) {
					$read[] = $obj->socket->get_socket();
				}
			}
			if (empty($read)) {
				break;
			}
			$n = socket_select($read, $write, $error, $t_sec, $t_usec);
			if ($n > 0) {
				// 可读
				foreach ($read as $sock) {
					$id = (int)$sock;
					$retObj = $this->wait_list[$id];
					$data = $retObj->socket->recv();
					// socket被关闭了
					if (empty($data)) {
						$retObj->code = SOA_Result::ERR_CLOSED;
						unset($this->wait_list[$id], $retObj->socket);
						continue;
					}
					// 一个新的请求，缓存区中没有数据
					if (!isset($buffer[$id])) {
						// 这里仅使用了length和type，uid,serid未使用
						$header[$id] = unpack(SOAServer::HEADER_STRUCT, substr($data, 0, SOAServer::HEADER_SIZE));
						// 错误的包头
						if ($header[$id] === false or $header[$id]['length'] <= 0) {
							$retObj->code = SOA_Result::ERR_HEADER;
							unset($this->wait_list[$id]);
							continue;
						} elseif ($header[$id]['length'] > $this->packet_maxlen) { // 错误的长度值
							$retObj->code = SOA_Result::ERR_TOOBIG;
							unset($this->wait_list[$id]);
							continue;
						}
						$buffer[$id] = substr($data, SOAServer::HEADER_SIZE);
					} else {
						$buffer[$id] .= $data;
					}
					// 达到规定的长度
					if (strlen($buffer[$id]) == $header[$id]['length']) {
						// 成功处理
						$this->finish(SOAServer::decode($buffer[$id], $header[$id]['type']), $retObj);
						$success_num++;
					}
					// 继续等待数据
				}
			}
			// 发生超时
			if ((microtime(true) - $st) > $timeout) {
				foreach ($this->wait_list as $obj) {
					$obj->code = ($obj->socket->connected) ? SOA_Result::ERR_TIMEOUT : SOA_Result::ERR_CONNECT;
				}
				// 清空当前列表
				$this->wait_list = array();
				return $success_num;
			}
		}
		// 未发生任何超时
		$this->wait_list = array();
		return $success_num;
	}
}
class SOA_Result {

	public $id;

	public $code = self::ERR_NO_READY;

	public $msg;

	public $data = null;

	public $send;
	// 要发送的数据
	public $type;

	/**
	 * 回调函数
	 * @var mixed
	 */
	public $callback;

	/**
	 * @var \Swoole\Client\TCP
	 */
	public $socket = null;

	const ERR_NO_READY = 8001;
	// 未就绪
	const ERR_CONNECT = 8002;
	// 连接服务器失败
	const ERR_TIMEOUT = 8003;
	// 服务器端超时
	const ERR_SEND = 8004;
	// 发送失败
	const ERR_SERVER = 8005;
	// server返回了错误码
	const ERR_UNPACK = 8006;
	// 解包失败了
	const ERR_HEADER = 8007;
	// 错误的协议头
	const ERR_TOOBIG = 8008;
	// 超过最大允许的长度
	const ERR_CLOSED = 8009; // 连接被关闭
}